package org.apache.spark.examples.sql.streaming;

import javafx.scene.input.DataFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import static org.apache.spark.sql.functions.col;

/**
 * Created by admin on 2019/3/21.
 */
public class Sparkmysql {
    // $example on:create_ds$
    public static class Person implements Serializable {
        private String name;
        private int age;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }
    }

    //hdfs目录
    private static final String filePath = "hdfs://172.30.17.164:8020/streamsets/people.json";
    private static final String filePathtxt = "hdfs://172.30.17.164:8020/streamsets/people.txt";
    static final String path = "/src/main/resources/people.json";

    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("LG_CALCULATE");
        conf.setMaster("local");
        System.setProperty("hadoop.home.dir", "C:/hadoop-2.6.0");
        SparkSession spark = SparkSession
                .builder()
                .appName("Java Spark SQL basic example")
                .config("spark.some.config.option", "some-value")
                .getOrCreate();
        try {
//            runBasicDataFrameExample(spark);
//            runDatasetCreationExample(spark);
//            runInferSchemaExample(spark);
            runProgrammaticSchemaExample(spark);
        } catch (Exception e) {
            e.printStackTrace();
        }
        spark.stop();
    }

    private static void runBasicDataFrameExample(SparkSession spark) throws AnalysisException {
//        Dataset<Row> df = spark.read().json(path);
        Dataset<Row> df = spark.read().json(filePath);
        df.show();
        System.out.println(df.collect());
        //以树格式
        df.printSchema();
        //仅选择“名称”列
        df.select("name");
        //选择所有人，但将年龄增加1
        df.select(col("name"), col("age").plus(1)).show();
        //选择年龄超过21
        df.filter(col("age").gt(21)).show();
        //计数按年龄的人
        df.groupBy("age").count().show();
        //将DataFrame注册为SQL临时视图
        df.createOrReplaceTempView("people");

        Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
        sqlDF.show();
        //将DataFrame注册为全局临时视图
        df.createGlobalTempView("people");
        //全局临时视图绑定到系统保留的数据库`global_temp`
        spark.sql("SELECT * FROM global_temp.people").show();
        //全局临时视图是跨会话
        spark.newSession().sql("SELECT * FROM global_temp.people").show();


    }

    private static void runDatasetCreationExample(SparkSession spark) {
        // $example on:create_ds$
        // Create an instance of a Bean class
        //创建Bean类的实例
        Person person = new Person();
        person.setName("Andy");
        person.setAge(32);
        //为Java bean创建
        // Encoders are created for Java beans
        Encoder<Person> personEncoder = Encoders.bean(Person.class);
        Dataset<Person> javaBeanDS = spark.createDataset(Collections.singletonList(person), personEncoder);
        javaBeanDS.show();
        // +---+----+
        // |age|name|
        // +---+----+
        // | 32|Andy|
        // +---+----+
        // Encoders
        // Encoders for most common types are provided in class Encoders
        Encoder<Integer> integerEncoder = Encoders.INT();
        Dataset<Integer> primitiveDS = spark.createDataset(Arrays.asList(1, 2, 3), integerEncoder);
        Dataset<Integer> transformedDS = primitiveDS.map(
                (MapFunction<Integer, Integer>) value -> value + 1,
                integerEncoder);
        transformedDS.collect(); // Returns [2, 3, 4]
        //通过提供类可以将DataFrame转换为数据集。基于名称的映射
        Dataset<Person> peopleDS = spark.read().json(filePath).as(personEncoder);
        peopleDS.show();
        // +----+-------+
        // | age|   name|
        // +----+-------+
        // |null|Michael|
        // |  30|   Andy|
        // |  19| Justin|
        // +----+-------+
        // $example off:create_ds$
    }

    /**
     * 使用反射推断模式
     *
     * @param spark
     */
    private static void runInferSchemaExample(SparkSession spark) {
        // $example on:schema_inferring$
        //从文本文件
        // Create an RDD of Person objects from a text file
        JavaRDD<Person> peopleRDD = spark.read()
                .textFile(filePathtxt)
                .javaRDD()
                .map(line -> {
                    String[] parts = line.split(",");
                    Person person = new Person();
                    person.setName(parts[0]);
                    person.setAge(Integer.parseInt(parts[1].trim()));
                    return person;
                });

//将模式应用于JavaBeans的RDD以获取DataFrame
        // Apply a schema to an RDD of JavaBeans to get a DataFrame
        Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
        // Register the DataFrame as a temporary view
        //将DataFrame注册为临时视图
        peopleDF.createOrReplaceTempView("people");
        //可以使用spark sql
        // SQL statements can be run by using the sql methods provided by spark
        Dataset<Row> teenagersDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19");

        // The columns of a row in the result can be accessed by field index
        Encoder<String> stringEncoder = Encoders.STRING();
        Dataset<String> teenagerNamesByIndexDF = teenagersDF.map(
                (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
                stringEncoder);
        teenagerNamesByIndexDF.show();
        // +------------+
        // |       value|
        // +------------+
        // |Name: Justin|
        // +------------+

        // or by field name
        Dataset<String> teenagerNamesByFieldDF = teenagersDF.map(
                (MapFunction<Row, String>) row -> "Name: " + row.<String>getAs("name"),
                stringEncoder);
        teenagerNamesByFieldDF.show();
        // +------------+
        // |       value|
        // +------------+
        // |Name: Justin|
        // +------------+
        // $example off:schema_inferring$
    }

    /**
     * 以编程方式指定架构
     * @param spark
     */
    private static void runProgrammaticSchemaExample(SparkSession spark) {
        // Create an RDD
        JavaRDD<String> peopleRDD = spark.sparkContext()
                .textFile(filePathtxt, 1)
                .toJavaRDD();

// The schema is encoded in a string
        String schemaString = "name age";

// Generate the schema based on the string of schema
        List<StructField> fields = new ArrayList<>();
        for (String fieldName : schemaString.split(" ")) {
            StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
            fields.add(field);
        }
        StructType schema = DataTypes.createStructType(fields);

// Convert records of the RDD (people) to Rows
        JavaRDD<Row> rowRDD = peopleRDD.map((Function<String, Row>) record -> {
            String[] attributes = record.split(",");
            return RowFactory.create(attributes[0], attributes[1].trim());
        });

// Apply the schema to the RDD
        Dataset<Row> peopleDataFrame = spark.createDataFrame(rowRDD, schema);

// Creates a temporary view using the DataFrame
        peopleDataFrame.createOrReplaceTempView("people");

// SQL can be run over a temporary view created using DataFrames
        Dataset<Row> results = spark.sql("SELECT name FROM people");

// The results of SQL queries are DataFrames and support all the normal RDD operations
// The columns of a row in the result can be accessed by field index or by field name
        Dataset<String> namesDS = results.map(
                (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
                Encoders.STRING());
        namesDS.show();
// +-------------+
// |        value|
// +-------------+
// |Name: Michael|
// |   Name: Andy|
// | Name: Justin|
// +-------------+

    }


}